package org.xserver.disruptor.core;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.Executor;

import javax.annotation.Resource;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Service;
import org.xserver.disruptor.dsl.stubs.StubExecutor;

import com.lmax.disruptor.BlockingWaitStrategy;
import com.lmax.disruptor.EventHandler;
import com.lmax.disruptor.ExceptionHandler;
import com.lmax.disruptor.dsl.Disruptor;
import com.lmax.disruptor.dsl.ProducerType;
/**
 * The <code>DisruptorEventFactory</code> to publis event
 * @author idoldefine
 * 
 */
@Service
public class DisruptorEventFactory implements EventFactory {
    private final Logger logger = LoggerFactory.getLogger(DisruptorEventFactory.class);
    @Resource
    private DisruptorParam disruptorParam;
    private static Map<Class, Disruptor> container = new HashMap<Class, Disruptor>();
    private static Map<Class, List<EventHandler>> handlerContainer = new HashMap<Class, List<EventHandler>>();

    public void post(Object event) {
        Class eventType = event.getClass();
        Disruptor disruptor = container.get(eventType);

        if (null != disruptor) {

            disruptor.publishEvent(DefaultEventFactory.getEventTranslator(event));
            logger.debug("Push event:{} success.", event);
        } else {
            logger.warn("EventType:{} not event consumer.", eventType);
        }
    }
   
      
    public void register(Object consumer, Class eventType) {
        EventHandler eventHandler = (EventHandler) consumer;
        if (container.containsKey(eventType)) {
            Disruptor disruptor = container.get(eventType);
            synchronized (disruptor) {
                destoryDisruptor(disruptor);
            }
            List<EventHandler> list = handlerContainer.get(eventType);
            list.add(eventHandler);
            handlerContainer.remove(eventType);

            createDisruptor(eventType, list);
        } else {
            List<EventHandler> list = new ArrayList<EventHandler>();
            list.add(eventHandler);
            createDisruptor(eventType, list);
        }
    }

    public void unregister(Object consumer, Class eventType) {
        if (container.containsKey(eventType)) {
            Disruptor disruptor = container.get(eventType);
            synchronized (disruptor) {
                disruptor.shutdown();
                container.remove(eventType);
            }

            //handler container process
            List<EventHandler> list = handlerContainer.get(eventType);
            list.remove(consumer);
            if (list.size() == 0) {
                handlerContainer.remove(eventType);
            } else {
                handlerContainer.put(eventType, list);
            }
        }
    }

    private void destoryDisruptor(Disruptor disruptor) {
        disruptor.shutdown();
    }

    private void createDisruptor(Class eventType, List<EventHandler> consumer) {
       StubExecutor executor = new StubExecutor();
//         Executor executor = Executors.newFixedThreadPool(disruptorParam.getThreadNum(), new ThreadFactory() {
//             public Thread newThread(Runnable r) {
//                 Thread t = new Thread(r);
//                 return t;
//             }
//         });
        createDisruptor(executor, eventType, consumer);
    }

    private void createDisruptor(final Executor executor, Class eventType, List<EventHandler> consumer) {
        Disruptor disruptor = new Disruptor(DefaultEventFactory.getEventFactory(eventType), disruptorParam.getRingBufferSize(), executor,
                ProducerType.SINGLE, new BlockingWaitStrategy());

        EventHandler[] eventHandlers = new EventHandler[consumer.size()];
        ExceptionHandler exceptionHandler = null;
        for (int i = 0; i < consumer.size(); i++) {
            eventHandlers[i] = consumer.get(i);
            if (consumer.get(i) instanceof ExceptionHandler) {
                exceptionHandler = (ExceptionHandler) consumer.get(i);
            }
        }
        //event consumer process
        disruptor.handleEventsWith(eventHandlers);

        //exception process
        if (null != exceptionHandler) {
            disruptor.handleExceptionsWith(exceptionHandler);
        } else {
            disruptor.handleExceptionsWith(new DisruptorExceptionHandler());
        }

        disruptor.start();
        container.put(eventType, disruptor);

        //handler container process
        handlerContainer.put(eventType, consumer);
    }

    public DisruptorParam getDisruptorParam() {
        return disruptorParam;
    }

    public void setDisruptorParam(DisruptorParam disruptorParam) {
        this.disruptorParam = disruptorParam;
    }
}
